
#include "logger.hpp"

#include "async_write.hpp"
#include "tablet_scanner.hpp"

namespace xbase
{
	
	void tablet_scanner::async_get_next_row(handler_type handler)
	{
		error_code ec;
		m_request_handler = handler;
		
		if (!m_started)
		{
			LOG(DEBUG, "tablet scan not started");

			// 扫描未开始，发送scan_open请求
			do_scan_open();
		}
		else if (m_scan_block.eof())
		{
			LOG(DEBUG, "tablet scan eof");

			// 已经没有更多记录，直接返回错误
			on_error(xbase::error::eos);
		}
		else if (m_scan_block.more())
		{
			LOG(DEBUG, "tablet scan block has more data");

			// 在内存中还有记录，直接返回
			uint64_t load_start_time = cur_micro_seconds();
			m_row_result_ptr->load(m_scan_block);
			m_scan_stat.m_local_process_time += (cur_micro_seconds() - load_start_time);

			++m_returned_row_count;
			dispatch(m_row_result_ptr, ec);
		}
		else
		{
			LOG(DEBUG, "tablet scan need to scan next");

			// 子表服务器还有更多记录，但内存中已经没有
			do_scan_next();
		}
	}

	void tablet_scanner::on_scan_open_response(uint64_t request_time,
						   protocol::packet_ptr pkt_ptr, const error_code &ec)
	{
		LOG(DEBUG, "scan_open_response ec=" << ec << " - " << ec.message());

		if (!ec)
		{
			scan_open_resp_type *p = boost::polymorphic_downcast<scan_open_resp_type*>(pkt_ptr.get());
			const protocol::scan_open_response &resp = p->body();
			const xbase::error::error_code resp_ec = resp.error_code();

			if(resp_ec)
			{
				if (resp_ec == error::eos)
				{
					dispatch(row_result_ptr(), resp_ec);
				}
				else if (resp_ec == xbase::error::tablet_closed)
				{
					on_scan_open_fail(resp_ec);
				}
				else
				{
					on_error(resp_ec);
				}
			}
			else
			{
				// 统计scan open信息
				m_scan_stat.m_scan_open_time += (cur_micro_seconds() - request_time);
				m_scan_stat.m_scan_open_times++;

				m_retry_times = 0;

				m_started = true;
				m_scanner_id = resp.m_scanner_id;

				m_scan_next_pkt_ptr = scan_next_request_ptr(new scan_next_request_type);
				m_scan_close_pkt_ptr = scan_close_request_ptr(new scan_close_request_type);

				// 初始化scan_next数据包
				protocol::scan_next_request &scan_next_pkt = m_scan_next_pkt_ptr->body();
				scan_next_pkt.m_tablet_name = m_scan_info.m_tablet_name;
				scan_next_pkt.m_scanner_id = m_scanner_id;

				// 初始化scan_close数据包
				protocol::scan_close_request &scan_close_pkt = m_scan_close_pkt_ptr->body();
				scan_close_pkt.m_tablet_name = m_scan_info.m_tablet_name;
				scan_close_pkt.m_scanner_id = m_scanner_id;

				do_scan_next();
			}
		}
		else
		{
			LOG(ERROR, "scan_open_response ec=" << ec << " - " << ec.message());
			// 网络错误，应该重试
			on_scan_open_fail(ec);
		}
	}

	void tablet_scanner::on_scan_next_response(uint64_t request_time,
						   protocol::packet_ptr pkt_ptr, const error_code &ec)
	{
		LOG(DEBUG, "scan_next_response ec=" << ec << " - " << ec.message());

		if (!ec)
		{
			scan_next_resp_type *p = boost::polymorphic_downcast<scan_next_resp_type*>(pkt_ptr.get());
			protocol::scan_next_response &resp = p->body();
			const xbase::error::error_code resp_ec = resp.error_code();

			if(resp_ec)
			{
				if (resp_ec == xbase::error::eos)
				{
					m_retry_times = 0;

					m_scan_block.load(resp);

					LOG(DEBUG, "record num: " << resp.m_records.size());

					if (m_scan_block.more())
					{
						uint64_t load_start_time = cur_micro_seconds();
						m_row_result_ptr->load(m_scan_block);
						m_scan_stat.m_local_process_time += (cur_micro_seconds() - load_start_time);

						++m_returned_row_count;
						dispatch(m_row_result_ptr, xbase::error::ok);
					}
					else
					{
						dispatch(row_result_ptr(), resp_ec);
					}

					do_scan_close();
				}
				else
				{
					on_error(resp_ec);
				}
			}
			else
			{
				// 统计scan next信息
				m_scan_stat.m_scan_next_time += (cur_micro_seconds() - request_time);
				m_scan_stat.m_scan_next_times++;

				m_retry_times = 0;

				m_scan_stat.m_scan_record_count += resp.m_records.size();
				LOG(DEBUG, "record num: " << resp.m_records.size());

				m_scan_block.load(resp);

				uint64_t load_start_time = cur_micro_seconds();
				m_row_result_ptr->load(m_scan_block);
				m_scan_stat.m_local_process_time += (cur_micro_seconds() - load_start_time);

				++m_returned_row_count;
				dispatch(m_row_result_ptr, xbase::error::ok);
			}
		}
		else
		{
			LOG(ERROR, "scan_next_response ec=" << ec << " - " << ec.message());
			// 需要重试
			on_scan_next_fail(ec);
		}
	}

	void tablet_scanner::on_scan_close_response(protocol::packet_ptr pkt_ptr, const error_code &ec)
	{
		if (!ec)
		{
			//std::cout << "recv resp : " << pkt_ptr->to_string() << std::endl;

			//scan_close_resp_type *p = boost::polymorphic_downcast<scan_close_resp_type*>(pkt_ptr.get());

			//const protocol::scan_close_response &resp = p->body();
		}
	}

	void tablet_scanner::do_scan_open()
	{
		xbase::async_write(m_session_manager, m_server_endpoint, m_scan_open_pkt_ptr,
				   boost::bind(&this_type::on_scan_open_response, this_ptr(), cur_micro_seconds(), _1, _2));
	}

	void tablet_scanner::do_scan_next()
	{
		xbase::async_write(m_session_manager, m_server_endpoint, m_scan_next_pkt_ptr,
				   boost::bind(&this_type::on_scan_next_response, this_ptr(), cur_micro_seconds(), _1, _2));
	}

	void tablet_scanner::do_scan_close()
	{
		xbase::async_write(m_session_manager, m_server_endpoint, m_scan_close_pkt_ptr,
				   boost::bind(&this_type::on_scan_close_response, this_ptr(), _1, _2));
	}

	void tablet_scanner::on_scan_open_fail(const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "tablet_scanner[" << m_scan_info.m_start_row
			    << "," << m_scan_info.m_end_row << "] scan_open retry: "
			    << ec << " - " << ec.message()
			    << ", retry_times = " << m_retry_times);

			++ m_retry_times;
			if(!m_timer)
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));

			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::do_scan_open, this_ptr()));
		}
		else
		{
			on_error(ec);
		}
	}

	void tablet_scanner::on_scan_next_fail(const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "tablet_scanner[" << m_scan_info.m_start_row
			    << "," << m_scan_info.m_end_row << "] scan_next retry: "
			    << ec << " - " << ec.message()
			    << ", retry_times = " << m_retry_times);


			++ m_retry_times;
			if(!m_timer)
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));

			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::do_scan_next, this_ptr()));
		}
		else
		{
			on_error(ec);
		}
	}

	void tablet_scanner::on_error(const error_code &ec)
	{
		LOG(ERROR, "tablet[" << m_scan_info.m_start_row << " , " << m_scan_info.m_end_row
		    << "] error: " << ec << " - " << ec.message()
		    << ", retry_times = " << m_retry_times);

		m_retry_times = 0;

		dispatch(row_result_ptr(), ec);
	}

	void tablet_scanner::dispatch(row_result_ptr row_ptr, const error_code &ec)
	{
		this->get_upper_level_service().dispatch(
			boost::bind(m_request_handler, row_ptr, ec)
			);
	}
}
